FastAPI নিরাপত্তা নিশ্চিত করার জন্য বিভিন্ন বৈশিষ্ট্য সরবরাহ করে। API-র নিরাপত্তা সুনিশ্চিত করার জন্য কিছু সাধারণ security best practices অনুসরণ করা গুরুত্বপূর্ণ। এই প্র্যাকটিসগুলোর মাধ্যমে আপনি সিস্টেমকে বিভিন্ন নিরাপত্তা হুমকি থেকে সুরক্ষিত রাখতে পারবেন।
এখানে FastAPI এর নিরাপত্তা সংক্রান্ত কিছু গুরুত্বপূর্ণ বিষয় এবং সেরা অভ্যাস আলোচনা করা হয়েছে।
1. CORS (Cross-Origin Resource Sharing) সেটআপ করা
CORS (Cross-Origin Resource Sharing) একটি নিরাপত্তা ফিচার যা ওয়েব ব্রাউজারে কনট্রোল করতে সহায়ক হয় কোন উৎস থেকে কনটেন্ট গ্রহণ করা যাবে। FastAPI তে CORS ব্যবস্থাপনা করতে CORSMiddleware ব্যবহার করা হয়।
উদাহরণ: CORS সেটআপ
from fastapi import FastAPI
from starlette.middleware.cors import CORSMiddleware
app = FastAPI()
# CORS Middleware যোগ করা
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # সকল ডোমেইনকে অনুমতি দিন (বিশেষভাবে প্রোডাকশন মোডে এটি নির্দিষ্ট ডোমেইনে সীমাবদ্ধ রাখতে হবে)
allow_credentials=True,
allow_methods=["*"], # সকল HTTP মেথড অনুমতি দিন
allow_headers=["*"], # সকল হেডার অনুমতি দিন
)
@app.get("/")
def read_root():
return {"message": "Hello, World!"}
এখানে:
allow_origins: কনফিগার করা হয় যে কোন উৎস থেকে API রিকোয়েস্ট গ্রহণ করা যাবে।allow_methods: কোন HTTP মেথড অনুমতি পাবে (যেমনGET,POST,PUT,DELETEইত্যাদি)।allow_headers: কোন হেডার অনুমতি পাবে।
উল্লেখ্য: প্রোডাকশন সার্ভারে allow_origins=["*"] ব্যবহার না করার পরামর্শ দেওয়া হয়, কারণ এটি সমস্ত উৎস থেকে অ্যাক্সেস দেয়। নিরাপত্তার জন্য আপনি নির্দিষ্ট ডোমেইন অ্যাড্রেস যুক্ত করতে পারেন।
2. Authentication এবং Authorization
API-র নিরাপত্তা নিশ্চিত করার জন্য authentication (ব্যবহারকারীর পরিচয় যাচাই) এবং authorization (ব্যবহারকারীর অধিকার যাচাই) খুবই গুরুত্বপূর্ণ। FastAPI-তে OAuth2 এবং JWT (JSON Web Token) ব্যবহার করে সুরক্ষিত অথেনটিকেশন বাস্তবায়ন করা যায়।
উদাহরণ: JWT Authentication সেটআপ
from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
from typing import List
import jwt
from datetime import datetime, timedelta
app = FastAPI()
# OAuth2PasswordBearer সেটআপ
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
# Secret Key
SECRET_KEY = "your_secret_key"
ALGORITHM = "HS256"
# JWT token জেনারেট করা
def create_access_token(data: dict, expires_delta: timedelta = timedelta(hours=1)):
to_encode = data.copy()
expire = datetime.utcnow() + expires_delta
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
# Token Validate
def verify_token(token: str = Depends(oauth2_scheme)):
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
return payload
except jwt.PyJWTError:
raise HTTPException(status_code=401, detail="Invalid token")
@app.post("/token")
def login(username: str, password: str):
# সাধারণত, এখানে আপনি ডাটাবেস বা কোনো সিস্টেম থেকে ব্যবহারকারীর তথ্য যাচাই করবেন
if username == "test" and password == "test":
return {"access_token": create_access_token(data={"sub": username}), "token_type": "bearer"}
raise HTTPException(status_code=401, detail="Incorrect username or password")
@app.get("/protected")
def protected_route(token: str = Depends(verify_token)):
return {"message": "This is a protected route", "user": token["sub"]}
এখানে:
- OAuth2PasswordBearer: একটি সিকিউরড অথেনটিকেশন স্কিমা (যেমন, Bearer Token) তৈরি করতে ব্যবহৃত হয়।
- JWT:
create_access_tokenফাংশনটি JWT টোকেন তৈরি করে এবংverify_tokenফাংশনটি টোকেনের বৈধতা যাচাই করে। tokenUrl: ইউজার লগইন করার জন্য একটি রাউট তৈরি করে।
3. HTTPS ব্যবহার করুন
HTTPS (Hypertext Transfer Protocol Secure) একটি নিরাপদ প্রটোকল যা সমস্ত ডাটা এনক্রিপ্ট করে এবং ম্যান-ইন-দ্য-মিডল (MITM) আক্রমণ থেকে সুরক্ষা প্রদান করে। আপনি যদি আপনার API-টিকে জনসাধারণের জন্য উপলব্ধ করেন, তবে HTTPS ব্যবহার করা অত্যন্ত গুরুত্বপূর্ণ।
uvicornএবংgunicorn-এর মাধ্যমে HTTPS সেটআপ করা সম্ভব।- SSL/TLS সার্টিফিকেট ব্যবহারের মাধ্যমে আপনার অ্যাপ্লিকেশন সুরক্ষিত করুন।
4. Input Validation এবং Data Sanitization
FastAPI ডাটার ভ্যালিডেশন এবং sanitization (দূষিত ডাটা পরিস্কার) জন্য Pydantic ব্যবহার করে। ফাইল, ইনপুট এবং প্যারামিটার যাচাইয়ের মাধ্যমে আপনি SQL ইনজেকশন এবং অন্যান্য নিরাপত্তা দুর্বলতা থেকে অ্যাপ্লিকেশনকে রক্ষা করতে পারেন।
উদাহরণ: Pydantic Validation
from pydantic import BaseModel, EmailStr
class User(BaseModel):
username: str
email: EmailStr
password: str
@app.post("/create_user/")
def create_user(user: User):
# এখানে ডাটা স্যানিটাইজ এবং নিরাপত্তা যাচাই করা হয়
return {"message": "User created successfully"}
এখানে, EmailStr ব্যবহার করে ইমেইল ফরম্যাট যাচাই করা হয়েছে এবং password ফিল্ডের জন্য স্যানিটাইজেশন প্রয়োগ করা হয়েছে।
5. Rate Limiting
Rate limiting ব্যবহারকারীকে একই রিকোয়েস্ট একাধিক বার পাঠানোর ক্ষেত্রে সীমাবদ্ধতা আরোপ করে। এটি DoS (Denial of Service) আক্রমণ প্রতিরোধে সাহায্য করে।
উদাহরণ: Rate Limiting
FastAPI তে slowapi লাইব্রেরি ব্যবহার করে rate limiting কনফিগার করা যায়।
pip install slowapi
from fastapi import FastAPI
from slowapi import Limiter
from slowapi.util import get_remote_address
app = FastAPI()
limiter = Limiter(key_func=get_remote_address)
@app.get("/limited-endpoint/")
@limiter.limit("5/minute")
def limited_endpoint():
return {"message": "You can access this endpoint only 5 times per minute."}
এখানে, @limiter.limit("5/minute") ব্যবহার করে প্রতি মিনিটে সর্বোচ্চ ৫টি রিকোয়েস্ট সীমাবদ্ধ করা হয়েছে।
6. SQL Injection থেকে সুরক্ষা
SQL ইনজেকশন আক্রমণ প্রতিরোধ করতে ORM (Object-Relational Mapping) ব্যবহার করা উচিত, যেমন SQLAlchemy। আপনি কখনও raw SQL কোড না লিখে ORM বা parameterized queries ব্যবহার করতে পারেন।
উদাহরণ: SQLAlchemy ORM ব্যবহার
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
email = Column(String, unique=True, index=True)
# Querying with ORM (avoiding raw SQL)
def get_user_by_email(db, email: str):
return db.query(User).filter(User.email == email).first()
এখানে, SQLAlchemy এর ORM ব্যবহার করে নিরাপদভাবে ডাটাবেস থেকে ডাটা নেওয়া হচ্ছে।
7. Logging and Monitoring
নিরাপত্তা নিশ্চিত করার জন্য logging এবং monitoring অত্যন্ত গুরুত্বপূর্ণ। আপনি loguru বা logging লাইব্রেরি ব্যবহার করে অ্যাপ্লিকেশনের কার্যক্রম লগ করতে পারেন।
উদাহরণ: Logging
import logging
from fastapi import FastAPI
app = FastAPI()
logging.basicConfig(level=logging.INFO)
@app.get("/")
def read_root():
logging.info("Root endpoint accessed.")
return {"message": "Hello, World!"}
এখানে, logging.info ব্যবহার করে info logs রাখা হয়েছে, যা অ্যাপ্লিকেশনের কার্যকলাপ ট্র্যাক করতে সাহায্য করবে।
FastAPI তে নিরাপত্তা নিশ্চিত করার জন্য CORS, Authentication, Authorization, Input Validation, Rate Limiting, SQL Injection Prevention, Logging, এবং HTTPS ব্যবহার করা অত্যন্ত গুরুত্বপূর্ণ। এই সব সেরা অভ্যাসগুলি অনুসরণ করে আপনি আপনার API-কে সুরক্ষিত রাখতে পারবেন এবং ডেটা নিরাপত্তা ও গোপনীয়তা নিশ্চিত করতে পারবেন।
FastAPI তে HTTPS এবং SSL/TLS কনফিগারেশন করা একটি গুরুত্বপূর্ণ পদক্ষেপ, বিশেষত যখন আপনার অ্যাপ্লিকেশন প্রোডাকশনে চলে আসে। HTTPS (HyperText Transfer Protocol Secure) ওয়েব ব্রাউজার এবং সার্ভারের মধ্যে নিরাপদ সংযোগ তৈরি করতে SSL/TLS প্রোটোকল ব্যবহার করে। এটি ডাটা এনক্রিপ্ট করে যাতে ট্রান্সমিশন সময়ে ডাটা চুরি বা ম্যানিপুলেশন হতে না পারে।
এই গাইডে, আমরা দেখব কিভাবে SSL/TLS কনফিগারেশন করে FastAPI অ্যাপ্লিকেশনকে HTTPS-এ রান করা যায়।
Step 1: SSL/TLS সার্টিফিকেট তৈরি করা
SSL সার্টিফিকেট এবং private key প্রাপ্ত করতে আপনাকে দুটি উপায় অবলম্বন করতে হবে:
- Self-signed certificate (যা সাধারণত ডেভেলপমেন্ট ও টেস্টিংয়ের জন্য ব্যবহৃত হয়)
- CA-signed certificate (যা প্রোডাকশনে ব্যবহৃত হয়)
1.1: Self-signed SSL সার্টিফিকেট তৈরি করা
OpenSSL ব্যবহার করে আপনি একটি self-signed সার্টিফিকেট তৈরি করতে পারেন।
- একটি টার্মিনাল খুলুন এবং নিচের কমান্ডটি চালান:
openssl genpkey -algorithm RSA -out server.key -aes256
এটি private key তৈরি করবে। তারপর SSL সার্টিফিকেট তৈরি করুন:
openssl req -new -key server.key -out server.csr
openssl x509 -req -in server.csr -signkey server.key -out server.crt
এখানে:
server.key: আপনার private key।server.crt: সার্টিফিকেট ফাইল।
এটি server.key এবং server.crt ফাইল দুটি তৈরি করবে, যা SSL কনফিগারেশনের জন্য প্রয়োজন।
1.2: CA-signed SSL সার্টিফিকেট
প্রোডাকশনে ব্যবহারের জন্য, আপনি একটি CA (Certificate Authority)-signed সার্টিফিকেট ব্যবহার করতে পারেন। এটি সাধারণত Let's Encrypt অথবা অন্যান্য SSL সার্ভিস প্রদানকারীর মাধ্যমে করা হয়।
Step 2: FastAPI অ্যাপ্লিকেশনে HTTPS কনফিগারেশন
এখন যে সার্টিফিকেট তৈরি করা হয়েছে তা FastAPI অ্যাপে কনফিগার করা হবে।
উদাহরণ: FastAPI অ্যাপ্লিকেশন HTTPS চালানো
FastAPI-তে HTTPS চালানোর জন্য uvicorn ব্যবহার করতে হবে। এটি --ssl-keyfile এবং --ssl-certfile প্যারামিটার ব্যবহার করে SSL/TLS কনফিগারেশন পরিচালনা করে।
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def read_root():
return {"message": "Hello, HTTPS!"}
এখন uvicorn সার্ভারটি ব্যবহার করে FastAPI অ্যাপ্লিকেশন চালাতে হবে SSL সার্টিফিকেট এবং প্রাইভেট কি ফাইল উল্লেখ করে।
uvicorn main:app --host 0.0.0.0 --port 443 --ssl-keyfile=server.key --ssl-certfile=server.crt
এখানে:
main:app: এটি অ্যাপ্লিকেশন ফাইলের নাম এবং FastAPI অ্যাপ অবজেক্টের নাম।--ssl-keyfile: আপনার private key ফাইলের পাথ।--ssl-certfile: আপনার SSL সার্টিফিকেট ফাইলের পাথ।
রেসপন্স:
এখন, আপনি ব্রাউজারে গিয়ে https://localhost অথবা https://yourdomain.com (আপনার ডোমেইনের উপর ভিত্তি করে) চেক করতে পারবেন যে FastAPI অ্যাপ HTTPS তে চলছে।
Step 3: HTTPS কনফিগারেশন প্রোডাকশনে (Reverse Proxy)
প্রোডাকশনে, আপনার FastAPI অ্যাপ্লিকেশন সাধারণত reverse proxy (যেমন Nginx বা Traefik) এর পিছনে চলে, যা SSL/TLS হ্যান্ডল করবে। আপনি যদি এমন একটি সেটআপে কাজ করেন, তাহলে SSL সার্টিফিকেটটি reverse proxy সার্ভারে কনফিগার করতে হবে এবং FastAPI অ্যাপ্লিকেশনটি HTTP-এ চলবে।
উদাহরণ: Nginx-এর সাথে SSL/TLS কনফিগারেশন
- Nginx Configuration File:
আপনার Nginx কনফিগারেশন ফাইলে নিম্নলিখিত সেটিংস যুক্ত করুন:
server {
listen 443 ssl;
server_name yourdomain.com;
ssl_certificate /path/to/your/certificate.crt;
ssl_certificate_key /path/to/your/private.key;
location / {
proxy_pass http://localhost:8000; # FastAPI অ্যাপ্লিকেশন চালু করার পোর্ট
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
}
}
server {
listen 80;
server_name yourdomain.com;
return 301 https://$host$request_uri;
}
এখানে:
- SSL সার্টিফিকেট এবং প্রাইভেট কী ফাইলগুলোর পাথ সঠিকভাবে উল্লেখ করুন।
- Proxy নির্দেশ করে FastAPI অ্যাপ্লিকেশনটির IP বা পোর্ট (যেমন:
localhost:8000), যা আপনি Uvicorn দিয়ে চালু করবেন।
- FastAPI অ্যাপ্লিকেশন চালানো:
FastAPI অ্যাপ্লিকেশনটি HTTP তে চলবে এবং Nginx reverse proxy দিয়ে HTTPS হ্যান্ডল করা হবে:
uvicorn main:app --host 127.0.0.1 --port 8000
এখন, HTTPS সংযোগটি Nginx-এর মাধ্যমে আপনার FastAPI অ্যাপ্লিকেশনটিতে রিডাইরেক্ট হবে।
Step 4: SSL Certificate Renewal (Let's Encrypt)
যদি আপনি Let's Encrypt থেকে সার্টিফিকেট ব্যবহার করেন, তাহলে আপনাকে Certbot ব্যবহার করে সার্টিফিকেট রিনিউ করতে হবে।
sudo certbot --nginx -d yourdomain.com
এটি Nginx সার্ভারের জন্য SSL সার্টিফিকেট রিনিউ এবং কনফিগার করবে।
FastAPI তে HTTPS এবং SSL/TLS কনফিগারেশন সেটআপ করা খুবই সহজ। আপনি নিজের self-signed certificate তৈরি করে ডেভেলপমেন্ট বা টেস্টিং করতে পারেন অথবা প্রোডাকশনে CA-signed certificate ব্যবহার করতে পারেন। FastAPI সার্ভারকে HTTPS মোডে চালাতে uvicorn ব্যবহার করা হয় এবং প্রোডাকশনে reverse proxy (যেমন Nginx) ব্যবহার করা হয় যাতে SSL/TLS হ্যান্ডল করা যায়। SSL সার্টিফিকেট ব্যবস্থাপনার জন্য Certbot ব্যবহার করা হয়।
API Rate Limiting এবং Throttling হল দুটি গুরুত্বপূর্ণ কৌশল যা API-এর কর্মক্ষমতা এবং সুরক্ষা বজায় রাখতে সাহায্য করে। এগুলি API তে অযাচিত বা অত্যধিক রিকোয়েস্ট ব্লক করতে এবং সার্ভারের অতিরিক্ত চাপ থেকে সুরক্ষা প্রদান করতে ব্যবহৃত হয়।
Rate Limiting হল একটি পদ্ধতি যা একটি নির্দিষ্ট সময়ের মধ্যে কতগুলি রিকোয়েস্ট একটি ব্যবহারকারী বা ক্লায়েন্ট থেকে গ্রহণ করা হবে তা সীমিত করে।
Throttling হল রিকোয়েস্টের পরিমাণ সীমাবদ্ধ করার একটি কৌশল যা সার্ভারের কর্মক্ষমতা এবং প্রসেসিং শক্তি বজায় রাখে, এবং নির্দিষ্ট সময়ের মধ্যে সর্বাধিক রিকোয়েস্ট সংখ্যা সীমাবদ্ধ করে।
FastAPI তে API Rate Limiting এবং Throttling কনফিগার করার জন্য আমরা বিভিন্ন পদ্ধতি ব্যবহার করতে পারি, যেমন Redis, Rate Limiting Libraries বা কাস্টম লজিক।
এখানে আমরা দেখব কিভাবে FastAPI তে API Rate Limiting এবং Throttling সেটআপ করা যায়।
Step 1: Rate Limiting Libraries ব্যবহার করা
FastAPI তে Rate Limiting এবং Throttling কনফিগার করার জন্য বেশ কিছু জনপ্রিয় লাইব্রেরি আছে। এর মধ্যে slowapi হল একটি খুবই সহজ এবং জনপ্রিয় লাইব্রেরি যা Redis-এর সাথে কাজ করে।
slowapi লাইব্রেরি ব্যবহার করা
slowapi লাইব্রেরি Redis-এর সাহায্যে FastAPI অ্যাপে Rate Limiting যোগ করতে সহায়ক। Redis ব্যবহার করা হলে, এটি বিভিন্ন রিকোয়েস্টের জন্য ভ্যালিডেশন এবং লিমিটেশন পরিচালনা করতে সাহায্য করে।
- প্রথমে
slowapiইনস্টল করুন:
pip install slowapi
pip install redis
- তারপরে FastAPI অ্যাপে এটি ব্যবহার করতে পারেন।
উদাহরণ: FastAPI তে Rate Limiting
from fastapi import FastAPI, Request
from slowapi import Limiter
from slowapi.util import get_remote_address
from fastapi.responses import JSONResponse
# Redis কনফিগারেশন
limiter = Limiter(key_func=get_remote_address)
app = FastAPI()
# Rate Limiting middleware
app.state.limiter = limiter
@app.get("/items/")
@limiter.limit("5/minute") # 5 রিকোয়েস্ট/মিনিট সীমা
async def get_items(request: Request):
return {"message": "This is a rate-limited endpoint!"}
@app.exception_handler(429)
async def ratelimit_error(request: Request, exc: Exception):
return JSONResponse(
status_code=429,
content={"detail": "Rate limit exceeded. Try again later."},
)
এখানে:
limiter.limit("5/minute"): এটি একটি পাথের জন্য প্রতি মিনিটে ৫টি রিকোয়েস্টের সীমা নির্ধারণ করে।- Redis ব্যাকএন্ডের মাধ্যমে এক্সপোনেনশিয়াল ব্যাকঅফ বা থ্রটলিংও বাস্তবায়ন করা যায়।
রেসপন্স (যদি Rate Limit অতিক্রম করা হয়):
{
"detail": "Rate limit exceeded. Try again later."
}
Step 2: Custom Rate Limiting এবং Throttling
আপনি যদি Redis ব্যবহার না করতে চান, তবে কাস্টম লজিক ব্যবহার করে FastAPI তে Rate Limiting এবং Throttling সেটআপ করতে পারেন।
উদাহরণ: Custom Rate Limiting Using in-memory Counter
from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
from collections import defaultdict
import time
app = FastAPI()
# Memory based rate limit tracking
rate_limits = defaultdict(list)
MAX_REQUESTS = 5 # Max requests per minute
TIME_FRAME = 60 # Time frame in seconds (1 minute)
@app.get("/limited/")
async def limited_endpoint(ip: str = None):
ip = ip or "127.0.0.1" # Using a fallback IP for simplicity
current_time = time.time()
request_times = rate_limits[ip]
# Remove requests that are older than the time window
rate_limits[ip] = [t for t in request_times if current_time - t < TIME_FRAME]
# Check if rate limit is exceeded
if len(rate_limits[ip]) >= MAX_REQUESTS:
raise HTTPException(
status_code=429,
detail="Rate limit exceeded. Try again later."
)
# Add the current request time
rate_limits[ip].append(current_time)
return {"message": "Request accepted!"}
এখানে:
rate_limits: এটি একটি ইন-মেমরি ডিকশনারি, যেখানে IP অ্যাড্রেস অনুযায়ী রিকোয়েস্টের টাইমস্ট্যাম্প রাখা হয়।MAX_REQUESTS = 5: প্রতি মিনিটে সর্বাধিক ৫টি রিকোয়েস্ট অনুমোদিত।TIME_FRAME = 60: রিকোয়েস্টের সময়কাল ১ মিনিট।
রেসপন্স (Rate Limit অতিক্রম করলে):
{
"detail": "Rate limit exceeded. Try again later."
}
Step 3: Exponential Backoff for Throttling
Throttling এর জন্য আপনি এক্সপোনেনশিয়াল ব্যাকঅফও বাস্তবায়ন করতে পারেন, যেখানে ব্যবহারকারী যদি অনেক বেশি রিকোয়েস্ট পাঠায়, তবে সময়সীমা বৃদ্ধি পাবে।
উদাহরণ: Exponential Backoff with Throttling
import time
from fastapi import FastAPI, HTTPException
app = FastAPI()
# Throttle settings
throttle_times = defaultdict(int)
@app.get("/throttled/")
async def throttled_endpoint(ip: str = None):
ip = ip or "127.0.0.1" # Using a fallback IP for simplicity
current_time = time.time()
# Throttling logic (increase wait time for consecutive requests)
throttle_times[ip] += 1
wait_time = 2 ** throttle_times[ip] # Exponential backoff: 2^n
if wait_time > 60: # Maximum wait time of 60 seconds
wait_time = 60
if time.time() - current_time < wait_time:
raise HTTPException(
status_code=429,
detail=f"Too many requests, try again in {wait_time} seconds."
)
return {"message": "Request accepted."}
এখানে, একাধিক রিকোয়েস্ট পাঠানোর পর ব্যবহৃত exponential backoff পদ্ধতিতে, পরবর্তী রিকোয়েস্টের জন্য অপেক্ষার সময় বাড়ানো হচ্ছে।
Step 4: Throttling Using Redis
Redis ব্যবহার করে Exponential Throttling কৌশল সহজেই বাস্তবায়ন করা সম্ভব। Redis-এ আপনি খুব সহজেই সময়ের ভিত্তিতে রিকোয়েস্টের সংখ্যা ট্র্যাক করতে পারেন এবং এক্সপোনেনশিয়াল ব্যাকঅফ থ্রটলিং সেটআপ করতে পারেন।
pip install redis
উদাহরণ: Redis এবং Throttling ব্যবহার
import redis
import time
from fastapi import FastAPI, HTTPException
app = FastAPI()
# Redis client setup
r = redis.StrictRedis(host='localhost', port=6379, db=0)
@app.get("/redis-throttle/")
async def redis_throttle(ip: str = None):
ip = ip or "127.0.0.1"
# Track requests in Redis (with expiration)
request_key = f"requests:{ip}"
current_time = time.time()
# Count requests for the IP
request_count = r.get(request_key)
if request_count and int(request_count) >= 5:
raise HTTPException(status_code=429, detail="Rate limit exceeded, try again later.")
# Increment the request count
r.incr(request_key)
r.expire(request_key, 60) # Expire the key after 1 minute
return {"message": "Request accepted!"}
এখানে, Redis ব্যবহার করে IP-based rate limiting এবং expiration বাস্তবায়ন করা হয়েছে।
FastAPI তে API Rate Limiting এবং Throttling সহজেই কনফিগার করা যায়। আপনি Redis, slowapi, অথবা কাস্টম লজিক ব্যবহার করে API রিকোয়েস্টের সংখ্যা নিয়ন্ত্রণ করতে পারেন। এর মাধ্যমে আপনি API-কে অযাচিত বা অত্যধিক লোড থেকে সুরক্ষা দিতে পারেন এবং সার্ভারের কর্মক্ষমতা বজায় রাখতে পারেন।
FastAPI একটি উচ্চ পারফরম্যান্স ওয়েব ফ্রেমওয়ার্ক, যা সিকিউরিটি ফিচারগুলোকে গুরুত্ব দিয়ে তৈরি করা হয়েছে। তবে, অ্যাপ্লিকেশন সিকিউরিটি নিশ্চিত করতে SQL Injection এবং Cross-Site Scripting (XSS) এর মত সাধারণ আক্রমণ থেকে রক্ষা পাওয়া খুবই গুরুত্বপূর্ণ।
এখানে আমরা আলোচনা করব কীভাবে FastAPI তে SQL Injection এবং XSS আক্রমণ থেকে রক্ষা পাওয়া যায়।
1. SQL Injection থেকে রক্ষা পাওয়া
SQL Injection হল একটি নিরাপত্তা দুর্বলতা, যেখানে আক্রমণকারী ক্ষতিকর SQL কোড ইনপুটের মাধ্যমে ডাটাবেসের তথ্য চুরি করতে বা নষ্ট করতে পারে। FastAPI তে SQLAlchemy বা Tortoise ORM ব্যবহার করে SQL Injection থেকে রক্ষা পাওয়া সহজ।
SQL Injection Prevention:
- Prepared Statements বা ORM ব্যবহার করুন: ORM ব্যবহার করার মাধ্যমে SQL Injection আক্রমণ প্রতিরোধ করা সম্ভব। ORM যেমন SQLAlchemy বা Tortoise ORM ইনপুট ভ্যালিডেশন এবং ডাটাবেস কুয়েরি নির্মাণে স্বয়ংক্রিয়ভাবে SQL Injection প্রতিরোধ করে।
- Parameterization: কুয়েরি প্যারামিটারাইজেশন ব্যবহার করুন। Direct SQL কুয়েরি লেখার সময়, প্যারামিটারাইজড কুয়েরি ব্যবহার করুন।
উদাহরণ: SQLAlchemy দিয়ে SQL Injection থেকে রক্ষা
from fastapi import FastAPI, HTTPException
from sqlalchemy.orm import Session
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# SQLAlchemy Setup
DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
app = FastAPI()
# প্যারামিটারাইজড কুয়েরি (Prepared Statements)
@app.get("/get_user/{user_id}")
def get_user(user_id: int, db: Session = Depends(get_db)):
user = db.query(User).filter(User.id == user_id).first() # SQL Injection প্রতিরোধ
if user is None:
raise HTTPException(status_code=404, detail="User not found")
return user
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
এখানে filter(User.id == user_id) প্যারামিটারাইজড কুয়েরি ব্যবহার করা হয়েছে, যাতে SQL Injection থেকে রক্ষা পাওয়া যায়। সরাসরি SQL কোড লিখলে, ইনপুটের মাধ্যমে SQL Injection হতে পারে, তবে ORM ব্যবহারে এই সমস্যা এড়ানো যায়।
SQL Injection Example (Unsafe Query)
# Unsafe direct SQL query
cursor.execute("SELECT * FROM users WHERE id = " + user_id)
এটি SQL Injection এর জন্য ঝুঁকিপূর্ণ। এটি ব্যবহার না করে ORM ব্যবহার করা উচিত।
2. XSS (Cross-Site Scripting) থেকে রক্ষা পাওয়া
Cross-Site Scripting (XSS) আক্রমণ হলো যেখানে আক্রমণকারী ক্ষতিকর স্ক্রিপ্ট ব্যবহার করে ইউজারের ব্রাউজারে ক্ষতিকর কোড চালায়, যা ইউজারের সেশন চুরি করতে বা ভুয়া ডাটা প্রদর্শন করতে ব্যবহৃত হতে পারে। FastAPI তে XSS থেকে রক্ষা পাওয়ার জন্য আপনি ইনপুট ভ্যালিডেশন এবং আউটপুট স্যানিটাইজেশন ব্যবহার করতে পারেন।
XSS Prevention:
- HTML Escaping: যখন HTML ডাটা রিটার্ন করবেন, তখন HTML escaping ব্যবহার করুন, যাতে স্ক্রিপ্ট ইনজেকশনের চেষ্টা ব্যর্থ হয়।
- Input Validation: ইউজারের ইনপুট ভ্যালিডেট করে নিশ্চিত করুন যে এতে কোন স্ক্রিপ্ট (JavaScript) কোড বা অপ্রত্যাশিত HTML কোড নেই।
- Use Templates Properly: FastAPI তে Jinja2 templates ব্যবহার করতে পারেন, যা ইনপুট ভ্যালিডেশন এবং HTML escaping স্বয়ংক্রিয়ভাবে করে দেয়।
উদাহরণ: Jinja2 Templates ব্যবহারের মাধ্যমে XSS থেকে রক্ষা
from fastapi import FastAPI, HTMLResponse
from fastapi.templating import Jinja2Templates
from starlette.requests import Request
app = FastAPI()
# Jinja2 Template Setup
templates = Jinja2Templates(directory="templates")
@app.get("/display_user/{user_name}", response_class=HTMLResponse)
async def display_user(request: Request, user_name: str):
return templates.TemplateResponse("user_profile.html", {"request": request, "user_name": user_name})
এখানে Jinja2Templates ব্যবহার করা হয়েছে, যা HTML escaping নিশ্চিত করে, যাতে ইউজারের ইনপুট যেমন "<script>alert('XSS')</script>" আক্রমণকৃত না হয়।
user_profile.html template:
<html>
<body>
<h1>Welcome, {{ user_name }}!</h1>
</body>
</html>
এখানে, Jinja2-এর {{ user_name }} সিনট্যাক্স স্বয়ংক্রিয়ভাবে HTML escaping করে, যাতে ইনপুটের মধ্যে যদি কোনো স্ক্রিপ্ট থাকে, তা চালানো না যায়।
Unsafe Example (XSS Vulnerability):
<html>
<body>
<h1>Welcome, <script>alert('XSS')</script></h1>
</body>
</html>
এখানে, <script>alert('XSS')</script> স্ক্রিপ্টটি XSS আক্রমণের জন্য ব্যবহার করা হতে পারে।
3. Security Headers ব্যবহার করা
FastAPI তে Security Headers যেমন Content-Security-Policy (CSP), X-Content-Type-Options, X-Frame-Options, এবং Strict-Transport-Security (HSTS) যোগ করা উচিত, যা XSS এবং অন্যান্য নিরাপত্তা দুর্বলতা থেকে সুরক্ষা দিতে সাহায্য করে।
উদাহরণ: Security Headers যোগ করা
from fastapi import FastAPI
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse
class SecurityMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
response = await call_next(request)
response.headers['Content-Security-Policy'] = "default-src 'self'"
response.headers['X-Content-Type-Options'] = "nosniff"
response.headers['X-Frame-Options'] = "DENY"
response.headers['Strict-Transport-Security'] = "max-age=63072000; includeSubDomains"
return response
app = FastAPI()
# Add Security Middleware
app.add_middleware(SecurityMiddleware)
@app.get("/")
async def main():
return {"message": "Security headers added!"}
এখানে, CSP, X-Content-Type-Options, X-Frame-Options, এবং HSTS সিকিউরিটি হেডারগুলোর মাধ্যমে XSS আক্রমণ প্রতিরোধ করা হয়েছে।
4. FastAPI তে Data Validation
FastAPI তে Pydantic ব্যবহার করে ইনপুট ভ্যালিডেশন করা হয়। আপনি ইনপুট ডাটার ধরণ এবং সীমাবদ্ধতা চেক করতে পারেন, যেমন কেবলমাত্র অনুমোদিত ক্যারেক্টার, সংখ্যা, বা টাইপ গ্রহণ করা উচিত।
উদাহরণ: ইনপুট ভ্যালিডেশন
from pydantic import BaseModel, constr
class Item(BaseModel):
name: constr(min_length=3, max_length=50) # Only allow strings with length 3-50
@app.post("/items/")
async def create_item(item: Item):
return {"item_name": item.name}
এখানে, name ফিল্ডের জন্য min_length এবং max_length সীমাবদ্ধতা দেওয়া হয়েছে, যাতে ইনপুট ভ্যালিডেশন করা যায়।
SQL Injection এবং XSS থেকে রক্ষা পেতে FastAPI শক্তিশালী input validation, HTML escaping, এবং ORM এর মাধ্যমে prepared statements ব্যবহার করে। Pydantic মডেল এবং Jinja2 templates সহ input sanitization এবং output encoding এর মাধ্যমে এই ধরনের আক্রমণ প্রতিরোধ করা সম্ভব। এছাড়া Security headers এবং CORS Middleware ব্যবহারের মাধ্যমে আপনি নিরাপত্তা আরও জোরালো করতে পারেন।
এই নিরাপত্তা ব্যবস্থাগুলোর মাধ্যমে, FastAPI তে আপনার অ্যাপ্লিকেশন নিরাপদ এবং আক্রমণ-মুক্ত রাখা সম্ভব।
Security Headers এবং Content Security Policy (CSP) একটি ওয়েব অ্যাপ্লিকেশনের সুরক্ষা নিশ্চিত করতে গুরুত্বপূর্ণ ভূমিকা পালন করে। ফাস্টএপিআই (FastAPI) তে সুরক্ষা নিশ্চিত করার জন্য বিভিন্ন ধরনের হেডার এবং CSP প্রয়োগ করা যেতে পারে। এই গাইডে আমরা Security Headers এবং CSP কীভাবে FastAPI তে ব্যবহার করা যায় তা আলোচনা করব।
১. Security Headers
Security Headers হল HTTP হেডার যা ব্রাউজারকে নির্দেশ দেয় কীভাবে অ্যাপ্লিকেশন সুরক্ষিতভাবে রেন্ডার হবে এবং কী ধরনের অপারেশনগুলো অনুমোদিত। এই হেডারগুলির মাধ্যমে আপনি আপনার অ্যাপ্লিকেশনের নিরাপত্তা শক্তিশালী করতে পারেন, যেমন Cross-Site Scripting (XSS), Cross-Site Request Forgery (CSRF), এবং অন্যান্য নিরাপত্তা ঝুঁকি থেকে রক্ষা করতে।
নিরাপত্তা হেডারের উদাহরণ
- Strict-Transport-Security (HSTS)
এটি ব্রাউজারকে বলবে যে শুধুমাত্র HTTPS প্রোটোকলের মাধ্যমে যোগাযোগ করা হবে। - X-Content-Type-Options
এটি ব্রাউজারকে বলে যে, কোন ফাইলের কনটেন্ট টাইপ ভুলভাবে অনুমোদিত হওয়া উচিত নয়। - X-Frame-Options
এটি কন্ট্রোল করে যে পেজটি iframe-এ এম্বেড করা যাবে কিনা। - X-XSS-Protection
এটি XSS আক্রমণ প্রতিরোধ করতে সাহায্য করে। - Content-Security-Policy (CSP)
এটি কন্ট্রোল করে কী ধরনের স্ক্রিপ্ট বা রিসোর্স আপনার পেজে লোড হবে।
২. FastAPI তে Security Headers যোগ করা
FastAPI তে Security Headers যোগ করতে আপনি Middleware ব্যবহার করতে পারেন। নিচে কিছু সাধারণ নিরাপত্তা হেডার যোগ করার উদাহরণ দেওয়া হলো:
উদাহরণ: Security Headers যোগ করা
from fastapi import FastAPI
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response
app = FastAPI()
class SecurityHeadersMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
response: Response = await call_next(request)
# Strict-Transport-Security (HSTS)
response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
# X-Content-Type-Options
response.headers["X-Content-Type-Options"] = "nosniff"
# X-Frame-Options
response.headers["X-Frame-Options"] = "DENY"
# X-XSS-Protection
response.headers["X-XSS-Protection"] = "1; mode=block"
# Content-Security-Policy (CSP)
response.headers["Content-Security-Policy"] = "default-src 'self'; img-src 'self'; script-src 'self';"
return response
# Security Middleware যুক্ত করা
app.add_middleware(SecurityHeadersMiddleware)
@app.get("/")
async def read_root():
return {"message": "FastAPI with Security Headers!"}
এখানে:
- Strict-Transport-Security (HSTS): HTTPS প্রোটোকল ব্যবহারের জন্য নির্ধারিত করা হয়েছে।
- X-Content-Type-Options: কনটেন্ট টাইপ সনাক্তকরণের ভুল ব্লক করতে।
- X-Frame-Options: ফ্রেমিংয়ের আক্রমণ (Clickjacking) থেকে রক্ষা করতে।
- X-XSS-Protection: XSS আক্রমণ প্রতিরোধ করতে।
- Content-Security-Policy (CSP): এই হেডারটি ব্রাউজারকে বলে দেয় কোন কনটেন্ট (যেমন স্ক্রিপ্ট, স্টাইল) কি উত্স (origin) থেকে লোড করা যাবে।
রেসপন্স:
Strict-Transport-Security: max-age=31536000; includeSubDomains
X-Content-Type-Options: nosniff
X-Frame-Options: DENY
X-XSS-Protection: 1; mode=block
Content-Security-Policy: default-src 'self'; img-src 'self'; script-src 'self';
৩. Content Security Policy (CSP)
Content Security Policy (CSP) হল একটি নিরাপত্তা ফিচার যা ওয়েব অ্যাপ্লিকেশনকে Cross-Site Scripting (XSS) এবং অন্যান্য ম্যালিসিয়াস আক্রমণ থেকে রক্ষা করতে ব্যবহৃত হয়। CSP হেডারটি বলে দেয় ব্রাউজারকে কোন উত্স থেকে কোন ধরনের কনটেন্ট লোড করা যাবে এবং কোন উত্স থেকে লোড করা যাবে না।
CSP হেডারের মূল উদ্দেশ্য:
- সক্রিপ্ট লোডিং নিয়ন্ত্রণ করা: শুধু নির্দিষ্ট উত্স থেকে স্ক্রিপ্ট লোড করার অনুমতি দেয়।
- স্টাইলশিট লোডিং নিয়ন্ত্রণ করা: নির্দিষ্ট উত্স থেকে CSS লোড করার অনুমতি দেয়।
- রিসোর্স লোডিং নিয়ন্ত্রণ করা: শুধু নির্দিষ্ট উত্স থেকে ছবি বা অন্য রিসোর্স লোড করার অনুমতি দেয়।
উদাহরণ: Content Security Policy সেট করা
from fastapi import FastAPI
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response
app = FastAPI()
class CSPMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
response: Response = await call_next(request)
# Content-Security-Policy
response.headers["Content-Security-Policy"] = (
"default-src 'self'; "
"img-src 'self'; "
"script-src 'self'; "
"style-src 'self';"
)
return response
# CSP Middleware যুক্ত করা
app.add_middleware(CSPMiddleware)
@app.get("/")
async def read_root():
return {"message": "FastAPI with Content Security Policy!"}
এখানে:
default-src 'self': সমস্ত কনটেন্ট শুধুমাত্র একই উত্স (যেমন,example.com) থেকে লোড হবে।img-src 'self': ইমেজ শুধুমাত্র একই উত্স থেকে লোড হবে।script-src 'self': স্ক্রিপ্ট শুধুমাত্র একই উত্স থেকে লোড হবে।style-src 'self': স্টাইলশিট শুধুমাত্র একই উত্স থেকে লোড হবে।
রেসপন্স:
Content-Security-Policy: default-src 'self'; img-src 'self'; script-src 'self'; style-src 'self';
৪. CSP এর অন্যান্য কনফিগারেশন
CSP হেডারটি কনফিগার করার সময় কিছু অতিরিক্ত কনফিগারেশন ব্যবহার করা যায়, যেমন:
report-uri: যদি কোনো CSP নিয়ম ভঙ্গ হয়, তাহলে একটি নির্দিষ্ট URL তে রিপোর্ট পাঠানো যাবে।unsafe-inline: ইন্টারনাল স্ক্রিপ্ট বা স্টাইলশিট অনুমোদন দেয় (যা নিরাপত্তার জন্য সাধারণত পরিহার করা হয়)।unsafe-eval: JavaScript কোডেeval()ব্যবহারের অনুমতি দেয়।
উদাহরণ: CSP এর রিপোর্টিং যুক্ত করা
response.headers["Content-Security-Policy"] = (
"default-src 'self'; "
"img-src 'self'; "
"script-src 'self'; "
"style-src 'self'; "
"report-uri /report-csp-violations;"
)
এখানে, যদি CSP নিয়ম ভঙ্গ হয়, তাহলে ব্রাউজার /report-csp-violations URL তে রিপোর্ট পাঠাবে।
Security Headers এবং Content Security Policy (CSP) হল ওয়েব অ্যাপ্লিকেশন সুরক্ষার জন্য অত্যন্ত গুরুত্বপূর্ণ। FastAPI তে এই হেডারগুলি ব্যবহার করে আপনি XSS, CSRF, Clickjacking এবং অন্যান্য নিরাপত্তা ঝুঁকি থেকে সুরক্ষা নিশ্চিত করতে পারেন। FastAPI তে সুরক্ষা হেডার এবং CSP সেট করা খুবই সহজ, এবং এগুলোর মাধ্যমে আপনি আপনার অ্যাপ্লিকেশনকে শক্তিশালী নিরাপত্তা ব্যবস্থা প্রদান করতে পারেন।
Read more
